package com.zhao.demo.unbound.demo04_transformation.sample04_sideoutput

import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

/**
 * Description: 使用侧输出流来优化之前的split的案例<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/8 17:25
 *
 * @author 柒柒
 * @version : 1.0
 */

object SideOutputStreamDemo {
  def main(args: Array[String]): Unit = {
    //1.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.实时读取流数据,计算,并展示结果
    import org.apache.flink.api.scala._

    //准备一个OutputTag的实例,即为:侧输出流中每个元素的标示值;注意:下述的泛型表示测输出流中每个元素的数据类型
    val outputTag: OutputTag[Raytek] = new OutputTag[Raytek]("异常体温")
    val totalStream: DataStream[Raytek] = env.socketTextStream("node01", 7775)
      .map(perInfo => {
        val arr = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).process(new ProcessFunction[Raytek, Raytek]() {

      /**
       * 处理流中的每一个元素,对应的业务场景是:红外体温测量仪监测每个即将跨过出口的旅客
       *
       * @param value
       * @param ctx
       * @param out
       */
      override def processElement(value: Raytek, ctx: ProcessFunction[Raytek, Raytek]#Context, out: Collector[Raytek]): Unit = {
        //判断
        if (value.temperature >= 36.3 && value.temperature <= 37.2) {
          //体温正常,使用实例Collector直接输出到主输出流中
          out.collect(value)
        } else {
          //体温异常,使用实例Context输出到侧输出流中
          ctx.output(outputTag, value)
        }
      }
    })

    //取出侧输出流中的旅客信息,进行处理
    totalStream.getSideOutput(outputTag)
      .print("当前的红外体温仪测到<体温异常>的旅客信息是->")

    //取出主输出流中的旅客信息,进行处理
    totalStream.print("当前的红外测温仪测到的<体温正常>的旅客信息是->")

    //3.启动
    env.execute()
  }
}













